import pymysql
import numpy as np
import pandas as pd
from sqlalchemy import create_engine
import time, datetime, os

if __name__ == '__main__':
    engine = create_engine('mysql+pymysql://5gzhyy:B6.5gzhyy312@132.91.175.98:8067/g41_wyzx_5gzhyydb')
    print(time.strftime(r"%Y-%m-%d %H:%M:%S", time.localtime()))

    from datetime import datetime, timedelta

    # 获取当前时间的前两天和前九天的日期字符串
    times1_str = (datetime.now() - timedelta(days=2)).strftime("%Y%m%d")
    times2_str = (datetime.now() - timedelta(days=9)).strftime("%Y%m%d")

    # times1_str = '20250621'   # 手动执行时最近周六日期
    # times2_str = '20250614'   # 手动执行时上个周六日期


    # 将字符串转换为 datetime 对象（但这里我们实际上只需要 times1）
    times1 = datetime.strptime(times1_str, "%Y%m%d")

    # 获取当前月份（注意：这里我们使用 times1 的月份）
    this_month_str = times1.strftime("%Y%m")

    # 找到本月的第一天
    first_day_of_month = times1.replace(day=1)

    # 计算给定日期是本月的第几天
    day_of_month = times1.day

    # 计算周数（简单方法，不是ISO周数）
    week_number = (day_of_month - 1) // 7 + 1

    # 上个月的第一天
    last_month_day_one = (first_day_of_month - timedelta(days=1)).replace(day=1)
    # 上个月的年份和月份（字符串格式）
    last_month_year_month = last_month_day_one.strftime("%Y%m")

    print(f"这个月: {this_month_str}")
    print(f"上个月: {last_month_year_month}")
    print(f"日期 {times1_str} 在本月的相对位置是第 {week_number} 周")

    # 打印前两天和前九天的日期（但注意，这里的前两天和前九天是基于 times1_str 的，而不是当前时间）
    print("前两天日期（根据指定日期计算）:",
          (datetime.strptime(times1_str, "%Y%m%d") - timedelta(days=2)).strftime("%Y%m%d"))
    print("前九天日期（根据指定日期计算）:",
          (datetime.strptime(times1_str, "%Y%m%d") - timedelta(days=9)).strftime("%Y%m%d"))


    file_path_1 = '/oss_luoshen/ywhx/SVIP_gz_table/tongji/'
    file_name_1 = 'svip8_tongji_'
    # 文件名使用转换后的日期字符串
    file_name = file_name_1 + times1_str + '.csv'  # 如果您想要使用原始字符串，这里就是这样
    # 或者，如果您想要使用基于 times1 但不同天数的日期，可以这样做：
    # file_name = file_name_1 + (times1 - timedelta(days=some_number)).strftime("%Y%m%d") + '.csv'
    print(f"文件名: {file_name}")

    sql_query = f"""
        select  {times1_str} as stat_date, {this_month_str} as month,
        {week_number} AS week_of_month,
        (select count(*) from svip8_yw_feedback where END_DATE = {times1_str}) as this_week_svipuser,
        (select count(distinct a.msisdn) from (select msisdn from svip8_yw_feedback where end_date = {times1_str}) a INNER JOIN (select msisdn from svip8_yw_fromkefu where end_date = {times2_str}) b ON a.msisdn = b.msisdn) as this_trace_svipuser,
        (select count(*)  from (select msisdn,
        CASE 
            WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
            WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
        END AS priority 
        from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a where a.priority=1) as priority1,
        (select count(*) from (select msisdn,
        CASE 
            WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
            WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
        END AS priority 
        from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a where a.priority=2) as priority2,
        (select count(*) from (select msisdn,
        CASE 
            WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
            WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
        END AS priority 
        from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a) as priorityall,	
        (select count(*) from (select msisdn,
        CASE 
            WHEN SUM(abnormal_events) = 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '2'
            WHEN SUM(abnormal_events) > 1 and SUM(abnormal_times) / NULLIF(SUM(call_times), 0) >= 0.01 THEN '1'
        END AS priority 
        from svip8_feed_to_wy where end_date = {times1_str} group by msisdn)a where a.priority=1) as this_poor_svipuser,
        (select count(distinct msisdn) from (select msisdn from svip8_yw_feedback where end_date = {times2_str} and abnormal_events >= 2 and abnormal_times / NULLIF(call_times, 0) >= 0.01 
            and msisdn in (select msisdn from svip8_yw_feedback where end_date = {times1_str}))a) as last_poo_svipuser_inthis,
        (select count(distinct msisdn) from (select msisdn from svip8_yw_feedback where end_date = {times2_str} and abnormal_events >= 2 and abnormal_times / NULLIF(call_times, 0) >= 0.01 
            and msisdn in (select msisdn from svip8_yw_feedback where end_date = {times1_str} and abnormal_events=0))a)  as last_poo_svipuser_better_inthis,
        (select count(*)  from svip8_feed_to_wy where end_date = {times1_str}) as this_poor_wy,
        (select count(distinct a.msisdn)  from svip8_feed_to_wy a where a.priority_week LIKE '%%/%%' and month = {last_month_year_month} and a.msisdn IN (select b.msisdn from svip8_yw_feedback b where end_date = {times1_str} GROUP BY b.msisdn)) as last_poor_wy_inthis,
        (select count(distinct a.msisdn) from (select msisdn from svip8_feed_to_wy where priority_week LIKE '%%/%%' and month = {last_month_year_month}) a join (select msisdn from svip8_yw_feedback where end_date = {times1_str} and ABNORMAL_EVENTS = 0) b on a.msisdn = b.msisdn) as last_poor_wy_better_inthis,
         (SELECT COUNT(DISTINCT a.msisdn) 
             FROM (SELECT msisdn FROM svip8_yw_feedback 
                   WHERE end_date = {times1_str} 
                   AND ((COALESCE(s_video_streaming_dl_lowspeed_times_02, 0) > 2) + 
                       (COALESCE(video_streaming_dl_lowspeed_times_02, 0) > 2) + 
                       (COALESCE(im_delay_extra_times, 0) > 2) +
                       (COALESCE(web_dl_low_speed_times, 0) > 2)) > 1) a) AS this_data_poor,
        (SELECT COUNT(DISTINCT a.msisdn)
         FROM (SELECT msisdn FROM svip8_yw_feedback
               WHERE end_date = {times2_str}
                 AND ((COALESCE(s_video_streaming_dl_lowspeed_times_02, 0) > 2) +
                     (COALESCE(video_streaming_dl_lowspeed_times_02, 0) > 2) +
                     (COALESCE(im_delay_extra_times, 0) > 2) +
                     (COALESCE(web_dl_low_speed_times, 0) > 2)) > 1 
                 AND msisdn IN (SELECT msisdn FROM svip8_yw_feedback WHERE end_date = {times1_str})) a) AS last_data_poor_inthis,
        (SELECT COUNT(DISTINCT msisdn) 
         FROM (SELECT msisdn FROM svip8_yw_feedback 
               WHERE end_date = {times2_str}
               AND ((COALESCE(s_video_streaming_dl_lowspeed_times_02, 0) > 2) +
                   (COALESCE(video_streaming_dl_lowspeed_times_02, 0) > 2) +
                   (COALESCE(im_delay_extra_times, 0) > 2) +
                   (COALESCE(web_dl_low_speed_times, 0) > 2)) > 1
               AND msisdn IN (
                   SELECT msisdn FROM svip8_yw_feedback 
                   WHERE end_date = {times1_str} 
                   AND ((COALESCE(s_video_streaming_dl_lowspeed_times_02, 0) > 2) +
                       (COALESCE(video_streaming_dl_lowspeed_times_02, 0) > 2) +
                       (COALESCE(im_delay_extra_times, 0) > 2) +
                       (COALESCE(web_dl_low_speed_times, 0) > 2)) <= 1)) a) AS last_data_poo_better_inthis
         ;"""

    df = pd.read_sql_query(sql_query, engine)
    df.to_sql('svip8_tongji', engine, if_exists='append', index=False, chunksize=100)

    if not df.empty:
        full_file_path = os.path.join(file_path_1, file_name)
        df.to_csv(full_file_path, index=False, encoding='utf-8')  # 使用 full_file_path
        os.chmod(full_file_path, 0o777)
        print("数据已成功追加到数据库。")
    else:
        print(f"DataFrame is empty for time " + times1_str + ", not exporting to CSV")